package com.mama100.rpc.mdp.job;

import java.math.BigDecimal;
import java.sql.Types;
import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import javax.annotation.Resource;

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;
import kafka.message.MessageAndMetadata;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.commons.lang3.time.FastDateFormat;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.dao.DataAccessException;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.SqlOutParameter;
import org.springframework.jdbc.core.SqlParameter;
import org.springframework.jdbc.core.simple.SimpleJdbcCall;
import org.springframework.stereotype.Service;

import com.google.common.collect.ImmutableList;
import com.google.common.collect.ImmutableMap;
import com.mama100.common.util.general.GlobalConfigUtil;
import com.mama100.coupon.inout.CouponConditionEnum;
import com.mama100.coupon.inout.CouponIssueInfo;
import com.mama100.rpc.mdp.service.ICouponService;

/**
 * 依赖数据库: <br>
 * 存储过程: SP_MDP_ISCATENEWCUST<br>
 * 表: MDP_COUPON_SEND_TASK ,MDP_COUPON_TRIGGER_PRODUCT_ID<br>
 * 序列:MDP_COUPON_SEND_TASK_SEQ,<br>
 * 
 * 
 * @author 10747
 */
@Service("couponKafkaJob")
public class CouponKafkaJob {

	public static String INSERT = "I";
	public static String UPDATE = "U";

	public static int ZEOR = 0;
	public static int SUCCESS = 1;
	public static int ERROR = 2;

	protected Logger log = LoggerFactory.getLogger(this.getClass());

	@Resource(name = "racJdbcTemplate")
	JdbcTemplate jdbcTemplate;

	@Resource
	private ICouponService icouponService;

	private Map<String, Thread> workThreads = new HashMap<String, Thread>();

	private FastDateFormat fdf = FastDateFormat.getInstance("yyyyMMdd");

	private List<String> procs;

	private List<String> triggerProductIds;

	public CouponKafkaJob() {

	}

	/**
	 * 工作线程 监听队列 "TX_PRODUCT_POINT_DETAILS." + yyyymmdd<br>
	 * 通过存储过程获取 couponDefId,couponConditionValue 然后发送优惠券.
	 *
	 */
	class WorkThread extends Thread {
		private String yyyymmdd;

		public WorkThread(String yyyymmdd) {
			super(yyyymmdd);
			this.yyyymmdd = yyyymmdd;

		}

		@Override
		public void run() {

			String topic = "TX_PRODUCT_POINT_DETAILS." + yyyymmdd;

			Properties props = GlobalConfigUtil.getInstance().getPropertiesByEnv("kafka.consumer.properties");
			ConsumerConfig config = new ConsumerConfig(props);

			ConsumerConnector comsumer = Consumer.createJavaConsumerConnector(config);

			Map<String, List<KafkaStream<byte[], byte[]>>> streams = comsumer.createMessageStreams(ImmutableMap.of(topic, 1));
			List<KafkaStream<byte[], byte[]>> streamList = streams.get(topic);

			KafkaStream<byte[], byte[]> stream = streamList.get(0);
			ConsumerIterator<byte[], byte[]> it = stream.iterator();

			// 存储过程 入参 为 积分id, 返回 优惠券id,优惠卷条件

			while (it.hasNext()) {
				comsumer.commitOffsets();

				MessageAndMetadata<byte[], byte[]> mm = it.next();
				String edit = new String(mm.message());

				String[] ss = StringUtils.splitPreserveAllTokens(StringUtils.remove(edit, "\""), "|");
				String op = ss[2];
				
				// 是插入 ,这里只监听插入!!! 因为1,11都是插入的!
				if (INSERT.equals(op)) {

					String id = ss[6];
					Long customerId = Long.parseLong(ss[7]);
					String productId = ss[8];
					String status = ss[11];
					// 是否在触发产品id里面
					if (triggerProductIds.contains(productId)) {
						log.info("--->" + edit);
						// 积分状态是成功
						if ("1".equals(status) || "11".equals(status)) {

							for (String proc : procs) {
								log.info("check if " + proc);
								SimpleJdbcCall call = new SimpleJdbcCall(jdbcTemplate).withProcedureName(proc).withSchemaName("mama100_owner");

								call.addDeclaredParameter(new SqlParameter("xid", Types.NUMERIC));
								call.addDeclaredParameter(new SqlParameter("xcustomer_id", Types.NUMERIC));
								call.addDeclaredParameter(new SqlOutParameter("couponDefId", Types.VARCHAR));
								call.addDeclaredParameter(new SqlOutParameter("couponConditionValue", Types.VARCHAR));

								Map<String, Object> ret = call.execute(id, customerId);
								
								String couponDefIdStr =  (String) ret.get("couponDefId");
								String couponConditionValueStr = (String) ret.get("couponConditionValue");
								log.info("return couponDefId:{},couponConditionValue:{}", couponDefIdStr, couponConditionValueStr);
								if (couponDefIdStr != null && couponConditionValueStr != null) {
									String[] couponDefIds = couponDefIdStr.split(",");
									String[] couponConditionValues = couponConditionValueStr.split(",");
									if(couponDefIds.length > 0 && couponDefIds.length == couponConditionValues.length){
										List<CouponIssueInfo> infos = new ArrayList<CouponIssueInfo>();
										
										for(int i=0;i<couponDefIds.length;i++){
											Long couponDefId = Long.parseLong(couponDefIds[i]);
											
											CouponIssueInfo info = new CouponIssueInfo();
											//优惠券定义ID
											info.setCouponDefId(couponDefId);
											//发行方式对应的值
											info.setCouponConditionValue(couponConditionValues[i]);
											//优惠券发行方式：1-送出,2-花钱买,3-积分兑,4-参加活动（枚举）
											info.setCouponCondition(CouponConditionEnum.CAMPAIGN);
											//平台渠道：-1-全部，1-在线支付，3、支付宝，4-微信支付
											//info.setPlatformChannel(platformChannel);
											
											infos.add(info);
										}
										
										try {
											//批量发送
											icouponService.sendbatchCoupin(customerId, infos);
											logDb(customerId, couponDefIdStr, couponConditionValueStr, SUCCESS, null);
										} catch (Throwable e) {
											log.error("send error", e);
											logDb(customerId, couponDefIdStr, couponConditionValueStr, ERROR, e.getMessage());
										}
									}
								}
							}
						}
					}
				}

			}
		}
	}

	/**
	 * 加载存储过程名称
	 */
	private void refreshProcs() {
		/*
		 * try { List<String> list = jdbcTemplate.queryForList(
		 * "select name from CPMS_COUPON_PROCS where enable =1 ", String.class);
		 * this.procs = list;
		 * 
		 * for (String s : list) { log.info("load proc:" + s); }
		 * 
		 * } catch (DataAccessException e) {
		 * log.error("refreshTriggerProductIds error", e); }
		 */
		this.procs = ImmutableList.of("SP_MDP_ISCATENEWCUST_BATCH");

	}

	/**
	 * 加载触发产品ID
	 */
	private void refreshTriggerProductIds() {

		try {
			List<String> list = jdbcTemplate.queryForList("select product_id from mama100_owner.MDP_COUPON_TRIGGER_PRODUCT_ID", String.class);
			this.triggerProductIds = list;

			for (String s : list) {
				log.info("load trigger product id:" + s);
			}

		} catch (DataAccessException e) {
			log.error("refreshTriggerProductIds error", e);
		}

	}

	private void stopWorkThread(String yyyymmdd) {
		Thread t = workThreads.get(yyyymmdd);
		if (t != null) {
			t.interrupt();
		}
		workThreads.remove(yyyymmdd);
	}

	public void startWorkThread(String yyyymmdd) {

		reloadDict();

		log.info("startWorkThread " + yyyymmdd);
		WorkThread t = new WorkThread(yyyymmdd);
		workThreads.put(yyyymmdd, t);
		t.start();
	}

	/**
	 * 启动明天的工作线程
	 */
	public void startTomorrowThread() {
		String yyyymmdd = fdf.format(DateUtils.addDays(new Date(), 1));
		startWorkThread(yyyymmdd);
	}

	/**
	 * 启动今天的工作线程
	 */
	public void startTodayThread() {

		String yyyymmdd = fdf.format(new Date());
		startWorkThread(yyyymmdd);

	}

	/**
	 * 停止昨天的工作线程
	 */
	public void stopYestodayThread() {

		Date yestoday = DateUtils.addDays(new Date(), -1);
		String yyyymmdd = fdf.format(yestoday);
		log.info("stopYestodayThread " + yyyymmdd);
		stopWorkThread(yyyymmdd);
	}

	/**
	 * 重新加载字典
	 */
	public void reloadDict() {
		refreshProcs();
		refreshTriggerProductIds();
	}

	/**
	 * 把发送记录写到数据库
	 * 
	 * @param customerId
	 * @param couponDefId
	 * @param couponConditionValue
	 * @param status
	 * @param retmsg
	 */
	private void logDb(Long customerId, String couponDefIdStr, String couponConditionValueStr, int status, String retmsg) {
		try {
			String sql = "insert into mama100_owner.MDP_COUPON_SEND_TASK(ID,CUSTOMER_ID,COUPON_DEF_ID_STR,COUPON_CONDIDION_VALUE_STR,SEND_DATE,STATUS,RETMSG) values(mama100_owner.MDP_COUPON_SEND_TASK_SEQ.nextval,?,?,?,?,?,?)";
			jdbcTemplate.update(sql, customerId, couponDefIdStr, couponConditionValueStr, new Date(), status, retmsg);
		} catch (DataAccessException e) {
			log.error("insert send record error", e);
		}
	}

}
